#include "MessageExchangeService_types.h"
#include "MessageExchangeService.h"

#include <iostream>
#include <boost/shared_ptr.hpp>
#include <boost/detail/atomic_count.hpp>

#include "melon/service/service_arch.h"
#include "melon/service/base_service_impl.h"
#include "base/logging.h"
#include "base/get_local_ip.h"
#include "base/config_reader.h"
#include "base/distributed_counter.h"

#include "im/common/common.h"

#include "rpc_thread_manager.h"
#include "create_message_task.h"
#include "bang_shopkeeper_cache.h"
#include "call_center_cache.h"

using namespace melon::service;

namespace imserver {

class MessageExchangeServiceHandler : public MessageExchangeServiceIf, public BaseServiceImpl {
 public:
  MessageExchangeServiceHandler() : distributed_counter_("/im_sequence_id"), sequence_step_(1000) {
  }
  virtual void CreateMessage(const ImMessage& message) {
    if (message.seq_id <= 0) {
      if (!cur_sequence_id_ || ((*cur_sequence_id_) % sequence_step_ == 0)) {
        int64_t read_id = distributed_counter_.GetAndInc(sequence_step_);
        cur_sequence_id_.reset(new boost::detail::atomic_count(read_id));
      }
      const_cast<ImMessage&>(message).__set_seq_id(++(*cur_sequence_id_));
      LOG_INFO("CreateMessage id=" << (*cur_sequence_id_) << " sender=" << message.sender);
    } else {
      // RECREATE_ON_LINGER_SESSION, 是为了遇到linger session时，给收发双方都有个提醒
      ImMessage & msg(const_cast<ImMessage&>(message));
      msg.flags &= ~MSG_FLAG_READ;  // 置为未读
      msg.flags |= MSG_FLAG_RECREATE_ON_LINGER;
      LOG_INFO("ReCreateMessage id=" << msg.seq_id << " sender=" << msg.sender);
    }

    using apache::thrift::concurrency::Runnable;
    GetThreadManager()->add(boost::shared_ptr<Runnable>(new CreateMessageTask(message)));
  }
 private:
  base::DistributedCounter distributed_counter_;
  const int64_t sequence_step_;
  boost::shared_ptr<boost::detail::atomic_count> cur_sequence_id_;
};

}

int main(int argc, char **argv) {
  char * conf_file = "../conf/message_exchange_service.conf";
  char opt;
  while ((opt = getopt(argc, argv, "c:")) != -1) {
    switch (opt) {
    case 'c':
      conf_file = optarg;
      break;
    default:
      std::cerr << "Unknown option " << optopt << std::endl;
      return 1;
    }
  }

  ConfigReader cfg(conf_file);
  if (!cfg) {
    std::cerr << "Config file " << conf_file << " read error!" << std::endl;
    return 1;
  }

  LOG_INIT(cfg.Get("log4cplus", "file"), cfg.Get("log4cplus", "level"));

  int port = cfg.GetWithType<int>("message_exchange_service", "port", 9092);
  LOG_INFO("listens on port " << port);

  using namespace imserver;
  BangShopKeeperCache::Instance().Reload();
  CallCenterCache::Instance().Reload();

  ThriftService<MessageExchangeServiceHandler, MessageExchangeServiceProcessor> service;
  service.StartRegister("/im/message_exchange_service", "1.0", 0, base::GetLocalIp(), port, 1);
  service.Start(port);
  return 0;
}

